package tech.xuwei.paimon.cdc;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class TableSync3 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        //创建用于输出的表
        String createInputDDL = "create table source_test1 \n" +
                "(\n" +
                "    id             bigint ,\n" +
                "    name  string     ,\n" +
                "    create_datetime       timestamp(0)   ,\n" +
                "    dt  string     ,\n" +
                "    PRIMARY KEY (`id`) NOT ENFORCED\n" +
                ")WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'localhost',\n" +
                "    'port' = '3320',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'database-name' = 'cdc_test',\n" +
                "    'table-name' = 'test1', \n" +
                "    'debezium.log.mining.strategy'='online_catalog'," +
                "    'debezium.database.tablename.case.insensitive'='false'\n" +
                "  " +

                ") ";
        tableEnv.executeSql(createInputDDL);
        tableEnv.executeSql("SHOW CREATE TABLE source_test1").print();

        //创建Paimon类型的Catalog
        tableEnv.executeSql(" CREATE CATALOG paimon_catalog WITH( " +
                " 'type'='paimon'," +
//                "'metastore' = 'hive '," +  // 默认是filesystem
                "'warehouse'='hdfs://localhost:9000/paimon'" +
                ")" +
                "");
        tableEnv.executeSql("USE CATALOG paimon_catalog");


        TableResult tableResult = tableEnv.executeSql("create table mysql_to_paimon_01 as  select * from `default_catalog`.`default_database`.`source_test1` ");

        tableResult.print();

        //向表中添加数据
        tableEnv.executeSql("select * from `paimon_catalog`.`default`.`mysql_to_paimon_01` ").print();


    }
}
